package org.apache.rocketmq.namesrv.routeinfo;

import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.protocol.header.namesrv.UnRegisterBrokerRequestHeader;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 批量注销broker服务
 */
public class BatchUnregistrationService extends ServiceThread {

    private final RouteInfoManager routeInfoManager;

    private final LinkedBlockingQueue<UnRegisterBrokerRequestHeader> unregistrationQueue;
    private static final Logger log = LoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);

    public BatchUnregistrationService(RouteInfoManager routeInfoManager,NamesrvConfig namesrvConfig) {
        this.routeInfoManager = routeInfoManager;
        this.unregistrationQueue = new LinkedBlockingQueue<>(namesrvConfig.getDefaultThreadPoolQueueCapacity());
    }

    public boolean submit(UnRegisterBrokerRequestHeader requestHeader) {
        return this.unregistrationQueue.offer(requestHeader);
    }
    @Override
    public String getServiceName() {
        return BatchUnregistrationService.class.getName();
    }

    @Override
    public void run() {
        while (thread.isAlive()){
            try {

                UnRegisterBrokerRequestHeader take = unregistrationQueue.take();
                final Set<UnRegisterBrokerRequestHeader> requestHeaderHashSet = new HashSet<>();
                //在这里把最开始取出来的请求也放到集合中
                unregistrationQueue.drainTo(requestHeaderHashSet);
                requestHeaderHashSet.add(take);
                if (!requestHeaderHashSet.isEmpty()) {
                    this.routeInfoManager.unRegisterBroker(requestHeaderHashSet);
                }
            } catch (Exception e) {
                log.error("BatchUnregistrationService run", e);
            }
        }

    }

    int queueLength() {
        return this.unregistrationQueue.size();
    }
}